package com.gitee.wsl.func.filter.bloomfilter


import com.gitee.wsl.ext.base.Arrays.fill
import com.gitee.wsl.ext.base.SIZE
import com.gitee.wsl.func.filter.bloomfilter.api.*
import com.gitee.wsl.func.filter.bloomfilter.api.CellExtractor.CellPredicate
import com.gitee.wsl.func.filter.bloomfilter.bean.BitMaps.getLongBit
import com.gitee.wsl.func.filter.bloomfilter.bean.BitMaps.numberOfBitMaps
import com.gitee.wsl.func.filter.bloomfilter.bean.Shape
import com.gitee.wsl.func.predicate.IntPredicate
import com.gitee.wsl.func.predicate.LongPredicate
import com.gitee.wsl.func.predicate.Predicate
import com.gitee.wsl.text.format.format


/**
 * A counting Bloom filter using an int array to track cells for each enabled bit.
 *
 *
 *
 * Any operation that results in negative counts or integer overflow of counts will mark this filter as invalid. This transition is not reversible. The
 * operation is completed in full, no exception is raised and the state is set to invalid. This allows the cells for the filter immediately prior to the
 * operation that created the invalid state to be recovered. See the documentation in [.isValid] for details.
 *
 *
 *
 *
 * All the operations in the filter assume the cells are currently valid, for example `cardinality` or `contains` operations. Behavior of an invalid
 * filter is undefined. It will no longer function identically to a standard Bloom filter that is the merge of all the Bloom filters that have been added to and
 * not later subtracted from the counting Bloom filter.
 *
 *
 *
 *
 * The maximum supported number of items that can be stored in the filter is limited by the maximum array size combined with the [Shape]. For example an
 * implementation using a [Shape] with a false-positive probability of 1e-6 and [Integer.MAX_VALUE] bits can reversibly store approximately 75
 * million items using 20 hash functions per item with a memory consumption of approximately 8 GB.
 *
 *
 * @see Shape
 *
 * @see CellExtractor
 *
 * @since 4.5.0-M1
 */
class ArrayCountingBloomFilter : CountingBloomFilter {
    /**
     * The shape of this Bloom filter.
     */
    override val shape: Shape

    /**
     * The cell for each bit index in the filter.
     */
    private val cells: IntArray

    /**
     * The state flag. This is a bitwise `OR` of the entire history of all updated
     * cells. If negative then a negative cell or integer overflow has occurred on
     * one or more cells in the history of the filter and the state is invalid.
     *
     *
     * Maintenance of this state flag is branch-free for improved performance. It
     * eliminates a conditional check for a negative cell during remove/subtract
     * operations and a conditional check for integer overflow during merge/add
     * operations.
     *
     *
     * Note: Integer overflow is unlikely in realistic usage scenarios. A cell
     * that overflows indicates that the number of items in the filter exceeds the
     * maximum possible size (number of bits) of any Bloom filter constrained by
     * integer indices. At this point the filter is most likely full (all bits are
     * non-zero) and thus useless.
     *
     *
     * Negative cells are a concern if the filter is used incorrectly by
     * removing an item that was never added. It is expected that a user of a
     * counting Bloom filter will not perform this action as it is a mistake.
     * Enabling an explicit recovery path for negative or overflow cells is a major
     * performance burden not deemed necessary for the unlikely scenarios when an
     * invalid state is created. Maintenance of the state flag is a concession to
     * flag improper use that should not have a major performance impact.
     */
    private var state = 0

    private constructor(source: ArrayCountingBloomFilter) {
        this.shape = source.shape
        this.state = source.state
        this.cells = source.cells.clone()
    }

    /**
     * Constructs an empty counting Bloom filter with the specified shape.
     *
     * @param shape the shape of the filter
     */
    constructor(shape: Shape) {
        this.shape = shape
        cells = IntArray(shape.numberOfBits)
    }

    override fun add(other: CellExtractor): Boolean {
        other.processCells(CellPredicate { idx: Int, addend: Int -> this.add(idx, addend) })
        return isValid
    }

    /**
     * Add to the cell for the bit index.
     *
     * @param idx the index
     * @param addend the amount to add
     * @return `true` always.
     */
    private fun add(idx: Int, addend: Int): Boolean {
        try {
            val updated = cells[idx] + addend
            state = state or updated
            cells[idx] = updated
            return true
        } catch (e: IndexOutOfBoundsException) {
            throw IllegalArgumentException(
                String.format("Filter only accepts values in the [0,%d) range", shape.numberOfBits), e
            )
        }
    }

    override fun asIndexArray(): IntArray {
        return (0..cells.size).filter({ i -> cells[i] > 0 }).toIntArray()
    }

    override fun cardinality(): Int {
        return (0..cells.size).count({ i -> cells[i] > 0 })
    }

    override fun characteristics(): Int {
        return BloomFilter.Companion.SPARSE
    }

    override fun clear() {
        fill(cells, 0)
    }

    override fun contains(bitMapExtractor: BitMapExtractor): Boolean {
        return contains(IndexExtractor.fromBitMapExtractor(bitMapExtractor!!))
    }

    override fun contains(indexExtractor: IndexExtractor): Boolean {
        return indexExtractor.processIndices(Predicate { idx: Int? -> cells[idx!!] != 0 })
    }

    /**
     * Creates a new instance of this [ArrayCountingBloomFilter] with the same properties as the current one.
     *
     * @return a copy of this BloomFilter.
     */
    override fun copy(): ArrayCountingBloomFilter {
        return ArrayCountingBloomFilter(this)
    }

    override val maxCell: Int
        get() = Int.Companion.MAX_VALUE

    override fun getMaxInsert(cellExtractor: CellExtractor): Int {
        val max = intArrayOf(Int.Companion.MAX_VALUE)
        cellExtractor.processCells(CellPredicate { x: Int, y: Int ->
            val count = cells[x] / y
            if (count < max[0]) {
                max[0] = count
            }
            max[0] > 0
        })
        return max[0]
    }


    override val isValid: Boolean
        /**
         * {@inheritDoc}
         *
         *
         *
         * *Implementation note*
         *
         *
         *
         *
         * The state transition to invalid is permanent.
         *
         *
         *
         *
         * This implementation does not correct negative cells to zero or integer overflow cells to [Integer.MAX_VALUE]. Thus the operation that generated
         * invalid cells can be reversed by using the complement of the original operation with the same Bloom filter. This will restore the cells to the state
         * prior to the invalid operation. Cells can then be extracted using [.processCells].
         *
         */
        get() = state >= 0

     override fun processBitMaps(predicate: LongPredicate): Boolean {
        val blocksm1 = numberOfBitMaps(cells.size) - 1
        var i = 0
        var value: Long
        // must break final block separate as the number of bits may not fall on the long boundary
        for (j in 0..<blocksm1) {
            value = 0
            for (k in 0..< Long.SIZE) {
                if (cells[i++] != 0) {
                    value = value or getLongBit(k)
                }
            }
            if (!predicate.test(value)) {
                return false
            }
        }
        // Final block
        value = 0
        var k = 0
        while (i < cells.size) {
            if (cells[i++] != 0) {
                value = value or getLongBit(k)
            }
            k++
        }
        return predicate.test(value)
    }

    override fun processCells(consumer: CellPredicate): Boolean {
        for (i in cells.indices) {
            if (cells[i] != 0 && !consumer.test(i, cells[i])) {
                return false
            }
        }
        return true
    }

     override fun processIndices(consumer: IntPredicate): Boolean {
        for (i in cells.indices) {
            if (cells[i] != 0 && !consumer.test(i)) {
                return false
            }
        }
        return true
    }

    override fun subtract(other: CellExtractor): Boolean {
        other.processCells(CellPredicate { idx: Int, subtrahend: Int -> this.subtract(idx, subtrahend) })
        return isValid
    }

    /**
     * Subtracts from the cell for the bit index.
     *
     * @param idx the index
     * @param subtrahend the amount to subtract
     * @return `true` always.
     */
    private fun subtract(idx: Int, subtrahend: Int): Boolean {
        try {
            val updated = cells[idx] - subtrahend
            state = state or updated
            cells[idx] = updated
            return true
        } catch (e: IndexOutOfBoundsException) {
            throw IllegalArgumentException(
                String.format("Filter only accepts values in the [0,%d) range", shape.numberOfBits), e
            )
        }
    }
}
